package edu.nepu.flink.api.state;

import edu.nepu.flink.api.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Date 2024/3/2 10:42
 * @Created by chenshuaijun
 */
public class ValueStateDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 这里将并行的设置为1,是方便不同key的数据进入到同一个分区
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> source = env.socketTextStream("hadoop102", 9999).map(new MapFunction<String, WaterSensor>() {

            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");
                return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
            @Override
            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                return element.getTs() * 1000;
            }
        }));
        /**
         * 我们主要的目标是监控同一个id的连续的两个vc值，如果他们之间的差值大于10，就要输出一条报警信息
         * 现在我们使用valueState就可以发现key隔离的现象，即使我们的并行度是1
         * 这个是我们测试所使用到的数据
         * s1,2,3
         * s1,4,14
         * s2,5,25
         * 下面是打印出来的报警信息
         * 当前的key为: s1当前的vc 14 之前的vc值为： 3 他们之间相差：11 所以产生了报警
         * 我们可以清晰的看出，valueState是可以实现key隔离的
         */
        source.keyBy(WaterSensor::getId).process(new KeyedProcessFunction<String, WaterSensor, String>() {
            ValueState<Integer> preValue;

            @Override
            public void open(Configuration parameters) throws Exception {
                preValue = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("pre-value", Types.INT));
            }

            @Override
            public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                if (preValue.value() == null){
                    preValue.update(value.vc);
                } else {
                    if (Math.abs(preValue.value() - value.vc) > 10) {
                        out.collect("当前的key为: "+value.id + "当前的vc "+value.getVc() + " 之前的vc值为： "+preValue.value() +" 他们之间相差：" + Math.abs(preValue.value() - value.vc)+" 所以产生了报警");
                    }
                    // 更新之前的值
                    preValue.update(value.vc);
                }
            }
        }).print();

        env.execute();
    }
}
